package com.nulldev.util.concurrency.threadIt.v4.executors.fje;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedTransferQueue;

import com.nulldev.util.JVM.JVM;
import com.nulldev.util.concurrency.threadIt.v4.executors.fje.interfaces.ThreadPicker;
import com.nulldev.util.concurrency.threadIt.v4.executors.fje.utils.CircularThreadPicker;
import com.nulldev.util.concurrency.threadIt.v4.executors.fje.utils.RoundRobinThreadPicker;

/**
 * {@link SWorkStealingThread} is the core implementation of
 * {@link ForkJoinThread}. It is based off of Doug Lea's "A Java Fork/Join
 * Framework" with a few tweaks.
 * 
 */
public class SWorkStealingThread extends ForkJoinThread {
	private final LinkedTransferQueue<AbstractForkJoinTask<?>> deq = new LinkedTransferQueue<AbstractForkJoinTask<?>>();
	private final ThreadPicker<ForkJoinThread> threadPicker = ForkJoinExecutorService.USE_RR_TP
			? new RoundRobinThreadPicker<ForkJoinThread>(service.getThreads())
			: new CircularThreadPicker<ForkJoinThread>(service.getThreads());

	private final int MAX_STEAL_ATTEMPTS;

	public SWorkStealingThread(ForkJoinExecutorService service) {
		super(service);
		this.MAX_STEAL_ATTEMPTS = service.getThreads().size() << 8;
	}

	@Override
	public void run() {
		service.getStopLatch().register();
		try {
			while (!(shutdownNow.get() || (shutdown.get() && deq.isEmpty()))) {
				AbstractForkJoinTask<?> task = deq.poll();

				if (task != null && !task.isDone()) {
					try {
						task.run();
					} catch (Throwable ex) {
						System.err.println(ex);
					}
					statistics.set("tasksRun", i -> i + 1);
				} else {
					steal(null);
				}
			}
		} finally {
			service.getStopLatch().arriveAndDeregister();
		}
	}

	@Override
	public <T> AbstractForkJoinTask<T> fork(AbstractForkJoinTask<T> task) {
		if (deq.tryTransfer(task)) {
			service.signalMoreWork();
			return task;
		} else if (deq.offer(task)) {
			return task;
		} else {
			JVM.errorln("[WorkStealingThread::fork] Failed to submit more work!");
			return task;
		}
	}

	@Override
	public <T> T join(AbstractForkJoinTask<T> task) {
		while (!task.isDone()) {
			AbstractForkJoinTask<?> t = deq.poll();
			if (t != null && !t.isDone()) {
				try {
					t.run();
				} catch (Throwable ex) {
					System.err.println(ex);
				}
				statistics.set("tasksRun", i -> i + 1);
			} else {
				steal(task);
			}
		}

		return task.getResult();
	}

	/**
	 * The steal method is executed when a {@link SWorkStealingThread} has no work
	 * on its deque. Tasks pulled from the back of other {@link ForkJoinThread}s'
	 * deques, in round-robin fashion, and executed, yielding for the owner thread.
	 * 
	 * If the {@link SWorkStealingThread} is waiting for another task to complete
	 * and that task finishes, the method will return.
	 * 
	 * After a number of attempted steals, the thread will go into a waiting state
	 * until another thread pushes work onto its deque.
	 * 
	 * @param waiting the task that {@link SWorkStealingThread} is waiting for
	 *                completion
	 */
	private void steal(AbstractForkJoinTask<?> waiting) {
		AbstractForkJoinTask<?> task = null;
		int stealAttempts = 0;

		do {
			try {
				if (waiting != null && waiting.isDone() || !deq.isEmpty()) {
					return;
				}

				if (waiting == null && ++stealAttempts >= MAX_STEAL_ATTEMPTS) {
					try {
						statistics.set("idleCount", i -> i + 1);
						service.awaitMoreWork();
						stealAttempts = 0;
					} catch (InterruptedException e) {
						return;
					}
				}

				Thread.yield();

				ForkJoinThread randomThread = (ForkJoinThread) threadPicker.nextThread();
				task = randomThread.getTask();

				if (task != null && !task.isDone()) {
					task.run();
					statistics.set("tasksStolen", i -> i + 1);
				}
			} catch (Throwable f) {
				if (!(f instanceof InterruptedException)) {
					System.err.println(f);
				} else
					throw f;
			}
		} while (!isInterrupted() && !service.isShutdown());
	}

	@Override
	protected AbstractForkJoinTask<?> getTask() {
		return this.deq.poll();
	}

	@Override
	protected void onShutdown() {
		interrupt();
	}

	@Override
	protected void onShutdownNow() {
		interrupt();
	}

	@Override
	protected List<AbstractForkJoinTask<?>> drainTasks() {
		final List<AbstractForkJoinTask<?>> drained = new ArrayList<AbstractForkJoinTask<?>>(deq.size());
		while (!deq.isEmpty()) {
			drained.add(deq.poll());
		}
		return drained;
	}
}